package com.shujia.stream

import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Duration, Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo1WC {
  def main(args: Array[String]): Unit = {

    //1、创建spark 环境
    var conf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("wc")

    val sc = new SparkContext(conf)


    //2、创建SparkStreaming的上下文对象
    /**
      * 需要指定batch的间隔时间，多久计算一次的意思
      * 时间越长延迟就越长
      * 时间越吞吐量会降低
      *
      */
    val ssc = new StreamingContext(sc, Durations.seconds(5))


    //3、读取数据
    //构建Dstream,  被动接受数据
    val ds: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.129.101", 8888)

    /**
      * reduceByKey ：指挥计算当前batch的数据
      *
      */

    //处理数据
    ds.flatMap(_.split(","))
      .map((_, 1))
      .reduceByKey(_ + _)
      .print() //实时打印出来，相当于action算子


    //启动任务
    ssc.start()
    //等待关闭
    ssc.awaitTermination()

  }
}
